package com.fanxuankai.boot.canal.client;

import com.alibaba.otter.canal.protocol.CanalEntry;
import com.fanxuankai.boot.canal.client.autoconfigure.CanalClientProperties;
import com.fanxuankai.boot.canal.client.autoconfigure.ConsumerConfig;
import com.fanxuankai.boot.canal.util.CommonUtils;
import com.fanxuankai.boot.canal.util.Topic;
import com.fanxuankai.boot.mqbroker.Event;
import com.fanxuankai.boot.mqbroker.producer.EventPublisher;

import javax.annotation.Resource;
import java.util.List;
import java.util.stream.Collectors;

/**
 * MQ 抽象消费者
 *
 * @author fanxuankai
 */
public class MqBrokerEntryConsumer implements EntryConsumer<List<Event<String>>> {
    @Resource
    private EventPublisher<String> eventPublisher;
    @Resource
    private CanalClientProperties properties;

    @Override
    public void accept(List<Event<String>> events) {
        eventPublisher.publish(events);
    }

    @Override
    public List<Event<String>> apply(EntryWrapper entryWrapper) {
        String topic = Topic.of(entryWrapper, properties);
        String schemaName = entryWrapper.getSchemaName();
        String tableName = entryWrapper.getTableName();
        ConsumerConfig consumerConfig = properties.getConsumerConfig(entryWrapper)
                .orElse(null);
        return entryWrapper.getAllRowDataList()
                .stream()
                .map(rowData -> {
                    Event<String> event = new Event<>();
                    event.setName(topic);
                    if (entryWrapper.getEventType() == CanalEntry.EventType.INSERT) {
                        event.setKey(CommonUtils.md5(rowData.getAfterColumnsList()));
                        event.setData(CommonUtils.jsonWithActualType(consumerConfig, rowData.getAfterColumnsList(),
                                schemaName, tableName, false));
                    } else if (entryWrapper.getEventType() == CanalEntry.EventType.UPDATE) {
                        event.setKey(CommonUtils.md5(rowData.getAfterColumnsList()));
                        event.setData(CommonUtils.jsonWithActualType(consumerConfig,
                                rowData.getBeforeColumnsList(),
                                rowData.getAfterColumnsList(), schemaName, tableName, false));
                    } else if (entryWrapper.getEventType() == CanalEntry.EventType.DELETE) {
                        event.setKey(CommonUtils.md5(rowData.getBeforeColumnsList()));
                        event.setData(CommonUtils.jsonWithActualType(consumerConfig, rowData.getBeforeColumnsList(),
                                schemaName, tableName, false));
                    }
                    return event;
                })
                .collect(Collectors.toList());
    }
}
